package cn._51doit
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import java.util.Properties
/**
 * ┏┓　　　┏┓
 * ┏┛┻━━━┛┻┓
 * ┃　　　　　　　┃ 　
 * ┃　　　━　　　┃
 * ┃　┳┛　┗┳　┃
 * ┃　　　　　　　┃
 * ┃　　　┻　　　┃
 * ┃　　　　　　　┃
 * ┗━┓　　　┏━┛
 *    　　┃　　　┃神兽保佑
 *    　　┃　　　┃代码无BUG！
 *    　　┃　　　┗━━━┓
 *    　　┃　　　　　　　┣┓
 *    　　┃　　　　　　　┏┛
 *    　　┗┓┓┏━┳┓┏┛
 *    　　　┃┫┫　┃┫┫
 *    　　　┗┻┛　┗┻┛
 *
 * @create: 2021-10-12 20:05
 * @author: 今晚打脑斧
 * @program: ProducerAckDemo
 * @Description:
 *              保证生产者写入数据是安全的
 *              设置acks为 1 或 -1/all
 *              同时还要设置retries大于0
 * 0(渣男)       :生产者向leader分区发送数据,但是不等待任何响应,就可以继续发送数据.
 * -1/all(暖男)  :不但要leader将数据落盘,还要等待所有的follower分区数据罗盘,然后响应leader分区,最后leader分区响应生产者,生产者才会继续写数据
 * 1            :就是生产者将数据发送给leader分区,leader分区将数据罗盘即响应producer
 *              ACKS如果等于-1/all最安全,但是效率低
 *              ACKS如果等于1相对安全, 效率相对较高
 *              ACKS如果等于0最不安全,但是效率最高
 **/
object ProducerAckDemo {
  def main(args: Array[String]): Unit = {
    //指定broker的地址
    val properties = new Properties
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092")
    //指定key序列化方式
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    properties.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //设置ACK
    properties.setProperty("acks", "1")
    //设置retries
    properties.setProperty("retries", Integer.MAX_VALUE.toString)

    //创建一个生产者
    val producer = new KafkaProducer[String, String](properties)

    val producerRecord = new ProducerRecord[String, String]("worldcount", null, "hello kafka")

    producer.send(producerRecord)

    producer.close()
  }

}
